package com.shujia.transformation

import org.apache.flink.api.common.functions.FlatMapFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

object Demo02FlatMap {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    // 传入一行数据 可能包含多个单词 之间用 逗号 分隔
    // 每个单词输出一行
    val linesDS: DataStream[String] = env.socketTextStream("master", 8888)

    // scala 风格
    linesDS.flatMap(line => line.split(","))

    // java 风格
    linesDS.flatMap(new FlatMapFunction[String, String] {
      // value 表示每一行数据
      // out 用于将数据 逐条 发出
      override def flatMap(value: String, out: Collector[String]): Unit = {
        val arr: Array[String] = value.split(",")
        for (elem <- arr) {
          out.collect(elem)
        }
      }
    }).print()


    env.execute()

  }

}
